package thread

import (
	"fmt"
	"time"

	"gitee.com/sansaniot/ssiot-core/facade/runtime"
	"gitee.com/sansaniot/ssiot-core/logger"

	"ssdevice/common/constant"
	"ssdevice/common/utils"
	"ssdevice/internal/device/models"
)

type DevVarDataThread struct {
}

func (DevVarDataThread) Run(threadName string) {
	logger.Infof("启动线程 - %s.", threadName)

	// 初始化子线程个数
	constant.VariableQueue = make([]*constant.CQueue, constant.DataRoutineChild)
	for i := 0; i < constant.DataRoutineChild; i++ {
		constant.VariableQueue[i] = constant.NewCQueue()
		go func(base int) {
			child := fmt.Sprintf("%s-%d", threadName, base)
			logger.Infof("启动子线程 - %s", child)
			q := constant.VariableQueue[base]
			dataList := make([]models.SysDevVariable, 0)
			for {
				time.Sleep(1 * time.Second)
				for {
					s := q.Dequeue()
					if s == nil {
						break
					}
					dataList = append(dataList, s.([]models.SysDevVariable)...)
					if len(dataList) > 5000 {
						// 批量处理数据
						dealDevVarData(dataList, child)
						// 清空(减少扩容)
						dataList = dataList[0:0]
					}
				}
				if len(dataList) > 0 {
					// 批量处理数据
					dealDevVarData(dataList, child)
					// 清空(减少扩容)
					dataList = dataList[0:0]
				}
			}
		}(i)
	}
}

func dealDevVarData(dataList []models.SysDevVariable, child string) {
	db := runtime.Runtime.GetDbByKey("db")
	if db == nil {
		return
	}
	upsertList := make([]models.SysDevVariable, 0)
	repeatKey := make([]string, 0)
	// 重复upsert键 取最新的
	for i := len(dataList) - 1; i >= 0; i-- {
		k := dataList[i].Sn + "#" + dataList[i].Key
		if utils.StrIn(k, repeatKey) {
			continue
		}
		upsertList = append(upsertList, dataList[i])
		repeatKey = append(repeatKey, k)
	}
	variableModel := &models.SysDevVariable{}
	variableModel.Orm = db
	t1 := time.Now()
	if err := variableModel.UpsertDoUpdate(&upsertList, []string{"sn", "key"}, []string{"value", "update_time"}); err != nil {
		logger.Error(err)
		return
	}
	logger.Infof("%s 变量表批量更新%d. 耗时%v", child, len(upsertList), time.Since(t1).Round(time.Millisecond))
}
